{
 "cells": [
  {
   "attachments": {},
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## NinaPro动作分类——卷积神经网络动作分类\n",
    "\n",
    "针对传统方法对大样本数据动作分类精度不高的问题，我们引入深度学习模型来对手势动作进行识别。卷积神经网络（Convolutional Neural Networks, CNN）是一类包含卷积计算且具有深度结构的前馈神经网络（Feedforward Neural Networks），是深度学习（deep learning）的代表算法之一，在近年来艺成功运用于诸多领域，比如图像分类、语音识别、视频分类等。自CNN出现，就不断有人尝试使用CNN识别NinaPro数据集中的首部运动。相比经典分类方法，卷积神经网络的分类精度远高于经典方法。"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "CNN_LSIT = [\"FCN\", \"NinaProNet\", \"Resnet18\"]\n",
    "CNN_METHOD = 2"
   ]
  },
  {
   "attachments": {},
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### 1.数据读取与预处理"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "from scipy import io\n",
    "import numpy as np\n",
    "\n",
    "def get_feature_dict(filename):\n",
    "    \"\"\"将ninapro_feature的MAT文件加载为字典\n",
    "\n",
    "    Args:\n",
    "        path: mat文件路径\n",
    "        filename: mat文件名\n",
    "\n",
    "    Returns:\n",
    "        数据集字典\n",
    "        [feat_set, featStim, featRep]\n",
    "    \"\"\"\n",
    "    # 读取MAT文件\n",
    "    print('load file: ' + filename + '...', end= '', flush=True)\n",
    "    dict_feature=io.loadmat(filename)\n",
    "    if (dict_feature != ()):\n",
    "        #print(ninapro_data.keys())\n",
    "        print('[ok]:%d'%(len(dict_feature['featStim'])), flush=True)\n",
    "    # 返回字典\n",
    "    return dict_feature\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "import numpy as np\n",
    "from matplotlib import pyplot as plt\n",
    "\n",
    "def merge_multisubject(b,e):\n",
    "    \"\"\"将多组数据从mat文件中提取出来，预处理后合并\n",
    "\n",
    "    Args:\n",
    "        b: 开始的受试者序号\n",
    "        e: 结束的受试者序号\n",
    "\n",
    "    Returns:\n",
    "        [emg,acc,gyro,mag,labels]肌电c12，加速度c36，角速度c36，磁强c36数据和标签。\n",
    "    \"\"\"\n",
    "    emg_feature = None\n",
    "    labels = None\n",
    "    # 去除0label\n",
    "    for i in range(b,e+1):\n",
    "        feature_dict = get_feature_dict(\"../feature/feature_S{0}.mat\".format(i))\n",
    "        index = []\n",
    "        for i in range(len(feature_dict['featStim'])):\n",
    "            if feature_dict['featStim'][i]!=0:\n",
    "                index.append(i)\n",
    "        # 重排元素\n",
    "        emg_temp = feature_dict['feat_set']\n",
    "        emg_temp = np.reshape(emg_temp,(-1,5,12))\n",
    "        emg_temp = np.swapaxes(emg_temp,1,2)\n",
    "        if(emg_feature is None):\n",
    "            emg_feature = emg_temp[index,:,:]\n",
    "            labels = feature_dict['featStim'][index,:]\n",
    "        else:\n",
    "            emg_feature = np.vstack((emg_feature,emg_temp[index,:,:])) \n",
    "            labels = np.vstack((labels,feature_dict['featStim'][index,:]))\n",
    "        #print('delete 0 label,',emg_temp[index,:,:].shape)\n",
    "    # 归一化\n",
    "    s = np.zeros(5)\n",
    "    for i in range(5):\n",
    "        s[i] = np.sum(np.abs(emg_feature[:,:,i]))/emg_feature[:,:,i].size\n",
    "        #print(\"avg=\",s)\n",
    "        if(s[i]!=0):\n",
    "            emg_feature[:,:,i] /= s[i]\n",
    "            emg_feature[:,:,i] -= 0.5*s[i]\n",
    "    return emg_feature,labels\n",
    "emg_feature,labels = merge_multisubject(1,4)\n",
    "print(emg_feature.shape)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# 检查数据\n",
    "import matplotlib.pyplot as plt\n",
    "plt.plot(labels)\n",
    "print(emg_feature[0])"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "#数据预处理\n",
    "def precess_data(feat,label):\n",
    "    \"\"\"将数据reshape到可以送入神经网络的size\n",
    "\n",
    "    Args:\n",
    "        feat: 特征序列\n",
    "        label: 标签\n",
    "\n",
    "    Returns:\n",
    "        [feat, label] 处理后的特征序列和标签\n",
    "    \"\"\"\n",
    "    if (CNN_LSIT[CNN_METHOD] == \"FCN\"):\n",
    "        feat = np.swapaxes(feat,1,2)\n",
    "        feat = np.expand_dims(feat,1)\n",
    "    if (CNN_LSIT[CNN_METHOD] == \"NinaProNet\"):\n",
    "        feat = np.swapaxes(feat,1,2)\n",
    "    if (CNN_LSIT[CNN_METHOD] == \"Resnet18\"):\n",
    "        feat = np.swapaxes(feat,1,2)\n",
    "        feat = np.expand_dims(feat,1)\n",
    "        \n",
    "    feat = feat.astype(np.float32)\n",
    "    label = label.flatten() - 1\n",
    "    return feat,label\n"
   ]
  },
  {
   "attachments": {},
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### 2. 模型结构"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "import torch\n",
    "import torch.nn as nn\n",
    "from torch.autograd import Variable\n",
    "from torchsummary import summary\n",
    "import torch.nn.functional as F\n",
    "from torchvision import models\n",
    "\n",
    "class NinaProNet(nn.Module):\n",
    "    def __init__(self, class_num=None, base_features=16, window_length=256, input_channels=10):\n",
    "        super(NinaProNet, self).__init__()\n",
    "        self.class_num = class_num\n",
    "        self.conv1 = nn.Sequential(\n",
    "            nn.Conv1d(in_channels=input_channels,\n",
    "                      out_channels=base_features * 2,\n",
    "                      kernel_size=3, stride=1, padding=1),\n",
    "            nn.BatchNorm1d(base_features * 2),\n",
    "            nn.ReLU(),\n",
    "            nn.Dropout(p=0.3),\n",
    "            nn.MaxPool1d(kernel_size=2, stride=2)\n",
    "        )\n",
    "        self.conv2 = nn.Sequential(\n",
    "            nn.Conv1d(in_channels=base_features * 2,\n",
    "                      out_channels=base_features * 4,\n",
    "                      kernel_size=3, stride=1, padding=1),\n",
    "            nn.BatchNorm1d(base_features * 4),\n",
    "            nn.ReLU(),\n",
    "            nn.Dropout(p=0.3),\n",
    "            nn.MaxPool1d(kernel_size=2, stride=2)\n",
    "        )\n",
    "        self.conv3 = nn.Sequential(\n",
    "            nn.Conv1d(in_channels=base_features * 4,\n",
    "                      out_channels=base_features * 4,\n",
    "                      kernel_size=3, stride=1, padding=1),\n",
    "            nn.BatchNorm1d(base_features * 4),\n",
    "            nn.ReLU(),\n",
    "            nn.Dropout(p=0.3),\n",
    "            nn.MaxPool1d(kernel_size=2, stride=2)\n",
    "        )\n",
    "        self.conv4 = nn.Sequential(\n",
    "            nn.Conv1d(in_channels=base_features * 4,\n",
    "                      out_channels=base_features * 4,\n",
    "                      kernel_size=3, stride=1, padding=1),\n",
    "            nn.BatchNorm1d(base_features * 4),\n",
    "            nn.ReLU(),\n",
    "            nn.Dropout(p=0.3),\n",
    "        )\n",
    "\n",
    "        self.mlp1 = nn.Sequential(\n",
    "            nn.Linear(base_features * 4 * int(window_length / 8), 256),\n",
    "            nn.ReLU(),\n",
    "            nn.Dropout(p=0.5),\n",
    "            nn.Linear(256, 100),\n",
    "            nn.ReLU(),\n",
    "            nn.Dropout(p=0.5)\n",
    "        )\n",
    "        self.mlp2 = nn.Linear(100, self.class_num)\n",
    "\n",
    "    def forward(self, x):\n",
    "        x = x.type(torch.cuda.FloatTensor)\n",
    "        x = self.conv1(x)\n",
    "        x = self.conv2(x)\n",
    "        x = self.conv3(x)\n",
    "        x = self.conv4(x)\n",
    "\n",
    "        x = self.mlp1(x.view(x.size(0), -1))\n",
    "        x = self.mlp2(x)\n",
    "        x = F.softmax(x, dim=1)\n",
    "        return x\n",
    "\n",
    "class FCN(nn.Module):\n",
    "    def __init__(self, input_size, class_num):\n",
    "        super().__init__()\n",
    "        self.class_num = class_num\n",
    "        self.input_size = input_size\n",
    "        self.fcn1 = nn.Sequential(\n",
    "            nn.Linear(in_features=input_size[0] * input_size[1], out_features=10000),\n",
    "            nn.Dropout(p=0.3),\n",
    "            nn.BatchNorm1d(10000),\n",
    "            nn.ReLU())\n",
    "        self.fcn2 = nn.Sequential(\n",
    "            nn.Linear(in_features=10000, out_features=1000),\n",
    "            nn.Dropout(p=0.3),\n",
    "            nn.BatchNorm1d(1000),\n",
    "            nn.ReLU())\n",
    "        self.fcn3 = nn.Sequential(\n",
    "            nn.Linear(in_features=1000, out_features=100),\n",
    "            nn.Dropout(p=0.3),\n",
    "            nn.BatchNorm1d(100),\n",
    "            nn.ReLU())\n",
    "        self.fcn4 = nn.Linear(in_features=100, out_features=self.class_num)\n",
    "\n",
    "    def forward(self, x):\n",
    "        x = x.type(torch.cuda.FloatTensor)\n",
    "        x = x.view(x.size(0), -1)\n",
    "        x = self.fcn1(x)\n",
    "        x = self.fcn2(x)\n",
    "        x = self.fcn3(x)\n",
    "        x = self.fcn4(x)\n",
    "        x = F.softmax(x, dim=1)\n",
    "        return x\n",
    "device = torch.device(\"cuda:0\" if torch.cuda.is_available() else \"cpu\")\n",
    "print(\"The model will be running on\", device, \"device\")\n",
    "\n",
    "model_test = None\n",
    "\n",
    "if (CNN_LSIT[CNN_METHOD] == \"FCN\"):\n",
    "    model_test = FCN(class_num=40,input_size=(5,12)).to(device)\n",
    "    summary(model_test,(1,5,12))\n",
    "\n",
    "if (CNN_LSIT[CNN_METHOD] == \"NinaProNet\"):\n",
    "    model_test = NinaProNet(class_num=40,input_channels=5,window_length=12).to(device)\n",
    "    summary(model_test,(5,12))\n",
    "\n",
    "if (CNN_LSIT[CNN_METHOD] == \"Resnet18\"):\n",
    "    model_test = models.resnet18(num_classes=40)\n",
    "    model_test.conv1 = nn.Conv2d(1, 64, kernel_size=7, stride=2, padding=3, bias=False)\n",
    "    model_test.to(device)\n",
    "    summary(model_test,(1,5,12))"
   ]
  },
  {
   "attachments": {},
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### 3. 模型训练"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "import time\n",
    "\n",
    "def testAccuracy(device, model, test_loader):\n",
    "    \n",
    "    model.eval()\n",
    "    accuracy = 0.0\n",
    "    total = 0.0\n",
    "    \n",
    "    with torch.no_grad():\n",
    "        for data in test_loader:\n",
    "            emg_data, labels = data\n",
    "            emg_data = Variable(emg_data.to(device))    # torch.Size([64, 1, 200, 12])\n",
    "            labels = Variable(labels.to(device))        # torch.Size([64])\n",
    "            # run the model on the test set to predict labels\n",
    "            outputs = model(emg_data)\n",
    "            # the label with the highest energy will be our prediction\n",
    "            _, predicted = torch.max(outputs.data, 1)\n",
    "            total += labels.size(0)\n",
    "            accuracy += (predicted == labels).sum().item()\n",
    "    \n",
    "    # compute the accuracy over all test images\n",
    "    accuracy = (100 * accuracy / total)\n",
    "    return(accuracy)\n",
    "\n",
    "train_accs = []\n",
    "train_loss = []\n",
    "test_accs = []\n",
    "def train(device, num_epochs, train_loader, test_loader):\n",
    "    best_accuracy = 0\n",
    "    model = model_test\n",
    "    loss_func = torch.nn.CrossEntropyLoss()\n",
    "    optimizer = torch.optim.Adam(model.parameters(), lr=1e-4)\n",
    "\n",
    "    \n",
    "    \n",
    "\n",
    "    for epoch in range(num_epochs):\n",
    "        running_loss = 0.0\n",
    "        for i,(inputs, labels) in enumerate(train_loader,0):#0是下标起始位置默认为0\n",
    "            # data 的格式[[inputs, labels]]       \n",
    "    #         inputs,labels = data\n",
    "            inputs = Variable(inputs.to(device))    # torch.Size([64, 1, 200, 12])\n",
    "            labels = Variable(labels.to(device))        # torch.Size([64]) \n",
    "            #初始为0，清除上个batch的梯度信息\n",
    "            optimizer.zero_grad()         \n",
    "\n",
    "            #前向+后向+优化     \n",
    "            outputs = model(inputs)\n",
    "            loss = loss_func(outputs,labels.long())\n",
    "            loss.backward()\n",
    "            optimizer.step()\n",
    "\n",
    "            # loss 的输出，每个一百个batch输出，平均的loss\n",
    "            running_loss += loss.item()\n",
    "            if i%100 == 99:\n",
    "                print('[%d,%5d] loss :%.3f' %\n",
    "                    (epoch+1,i+1,running_loss/100),end='',flush=True)\n",
    "                running_loss = 0.0\n",
    "            train_loss.append(loss.item())\n",
    "\n",
    "            # 训练曲线的绘制 一个batch中的准确率\n",
    "            correct = 0\n",
    "            total = 0\n",
    "            _, predicted = torch.max(outputs.data, 1)\n",
    "            total = labels.size(0)# labels 的长度\n",
    "            correct = (predicted == labels).sum().item() # 预测正确的数目\n",
    "            train_accs.append(100*correct/total)\n",
    "            if i%100 == 99:\n",
    "                print(' acc=%d'%(100*correct/total))\n",
    "            \n",
    "        accuracy = testAccuracy(device, model, test_loader)\n",
    "        print('For epoch', epoch+1,'the test accuracy over the whole test set is %d %%' % (accuracy))\n",
    "        # we want to save the model if the accuracy is the best\n",
    "        if accuracy > best_accuracy:\n",
    "            best_accuracy = accuracy\n",
    "            torch.save(model.state_dict(), \"../model/best_epoch{1}_{0}.pth\".format(epoch, (int(time.time())%1000000)))\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# 矩阵数据预处理\n",
    "# 训练集1-16\n",
    "emg_feature,labels = merge_multisubject(1,4)\n",
    "emg_feature,labels = precess_data(emg_feature,labels)\n",
    "print('train reshaped data:',emg_feature.shape)\n",
    "emg_feature_t = torch.tensor(emg_feature)\n",
    "labels_t = torch.tensor(labels)\n",
    "train_dataset = torch.utils.data.TensorDataset(emg_feature_t, labels_t)\n",
    "\n",
    "# 验证集17-20\n",
    "emg_feature,labels = merge_multisubject(5,5)\n",
    "emg_feature,labels = precess_data(emg_feature,labels)\n",
    "print('test reshaped data:',emg_feature.shape)\n",
    "emg_feature_t = torch.tensor(emg_feature)\n",
    "labels_t = torch.tensor(labels)\n",
    "test_dataset = torch.utils.data.TensorDataset(emg_feature_t, labels_t)\n",
    "\n",
    "print('get dataloader...', end='',flush=True)\n",
    "# 划分数据集与训练集\n",
    "train_loader = torch.utils.data.DataLoader(\n",
    "    dataset=train_dataset,      # torch TensorDataset format\n",
    "    batch_size=128,      # mini batch size\n",
    "    shuffle=True,               # 要不要打乱数据 (打乱比较好)\n",
    "    num_workers=2,              # 多线程来读数据\n",
    "    drop_last = True,\n",
    "   \n",
    ")\n",
    "test_loader = torch.utils.data.DataLoader(\n",
    "    dataset=test_dataset,      # torch TensorDataset format\n",
    "    batch_size=256,      # mini batch size\n",
    "    shuffle=True,               # 要不要打乱数据 (打乱比较好)\n",
    "    num_workers=2,              # 多线程来读数据\n",
    "    drop_last = True,\n",
    ")\n",
    "print('[ok]')\n",
    "print('begin to train.....')\n",
    "# 模型训练\n",
    "train(device, 10, train_loader, test_loader)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "train_loss = np.array(train_loss)\n",
    "train_accs = np.array(train_accs)\n",
    "plt.plot(train_accs)\n",
    "plt.plot(train_loss)\n",
    "np.savetxt('train_loss.csv', train_loss, delimiter=\",\")\n",
    "np.savetxt('train_accs.csv', train_accs, delimiter=\",\")\n"
   ]
  },
  {
   "attachments": {},
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### 4. 模型验证"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "#生成验证集\n",
    "d,l = merge_multisubject(9,9)\n",
    "d,l = precess_data(d,l)\n",
    "print('reshaped data:',d.shape)\n",
    "\n",
    "print('get dataloader...', end='',flush=True)\n",
    "d_t = torch.tensor(d)\n",
    "l_t = torch.tensor(l)\n",
    "vaild_dataset = torch.utils.data.TensorDataset(d_t, l_t)\n",
    "vaild_loader = torch.utils.data.DataLoader(\n",
    "    dataset=vaild_dataset,      # torch TensorDataset format\n",
    "    batch_size=256,      # mini batch size\n",
    "    shuffle=False,               # 要不要打乱数据 (打乱比较好)\n",
    "    num_workers=2,              # 多线程来读数据\n",
    "    drop_last = True,\n",
    ")\n",
    "print('[ok]')\n",
    "\n",
    "model_test = models.resnet18(num_classes=40)\n",
    "model_test.conv1 = nn.Conv2d(1, 64, kernel_size=7, stride=2, padding=3, bias=False)\n",
    "model_test.load_state_dict(torch.load(\"../model/best_epoch849085_4.pth\"))\n",
    "model_test = model_test.to(device)\n",
    "accuracy = testAccuracy(device, model_test, vaild_loader)\n",
    "print('the test accuracy over the whole test set is %d %%' % (accuracy))"
   ]
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": "torch_env",
   "language": "python",
   "name": "python3"
  },
  "language_info": {
   "codemirror_mode": {
    "name": "ipython",
    "version": 3
   },
   "file_extension": ".py",
   "mimetype": "text/x-python",
   "name": "python",
   "nbconvert_exporter": "python",
   "pygments_lexer": "ipython3",
   "version": "3.8.13 (default, Oct 19 2022, 22:38:03) [MSC v.1916 64 bit (AMD64)]"
  },
  "vscode": {
   "interpreter": {
    "hash": "91eaa33755ee0e6c8927d7837736bb7bf44cb27baec87b08c7a4a0a98fc82110"
   }
  }
 },
 "nbformat": 4,
 "nbformat_minor": 4
}
